package com.ausware.studyforspark02;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import org.apache.commons.collections.IteratorUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.mortbay.log.Log;

import com.alibaba.fastjson.JSON;
import com.ausware.pojo.Mysql;
import com.ausware.pojo.ParseJson;
import com.ausware.pojo.Person;
import com.ausware.pojo.PersonTo;

public class Demo01 {
	@SuppressWarnings({ "resource", "unchecked", "rawtypes" })
	public static void main(String[] args) throws InterruptedException {
		SparkConf conf = new SparkConf().setMaster("local").setAppName("cql");
		JavaSparkContext jsc = new JavaSparkContext(conf);
		JavaRDD<String> input = jsc.textFile("src/main/resources/work.txt");
		JavaRDD<Person> result = input.mapPartitions(new ParseJson());
		Long time = System.currentTimeMillis();
		//Log.warn(time.toString());
		System.out.println(time);
		JavaPairRDD<String, Iterable<Person>> jp = (JavaPairRDD) result.groupBy(x -> x.getSessionId());
		JavaRDD<PersonTo> jpt = jp.map(x -> {
			PersonTo pt = new PersonTo();
			Iterator<Person> it = x._2.iterator();
			List<Person> myList = IteratorUtils.toList(it);
			List<Person> pickList = new LinkedList<>();
			for (Person p : myList) {
				if (time - p.getTimeMillis() <= 5 * 60 * 1000) {
					pickList.add(p);
					}
					for(Person per : pickList){
					//myList.remove(p);
					pt.setSessionId(p.getSessionId());
					pt.setContent(pt.getContent() + p.getContent());
					pt.setEnd(p.getEnd());
					if (pickList.size() == 1) {
						pt.setStartTimeMillis(p.getTimeMillis());
						pt.setEndTimeMillis(p.getTimeMillis());

					} else {
						pt.setStartTimeMillis(pickList.get(0).getTimeMillis());
						pt.setEndTimeMillis(pickList.get(pickList.size() - 1).getTimeMillis());
					}
				}
					pickList.clear();
			}
			
			return pt;
		});

		// Thread.sleep(5 * 1000);

		jpt.foreach(x -> {
			if (!(x.getSessionId() == null || x.getSessionId() == "")) {
				// 转为String的JSON格式数据
				String str = JSON.toJSONString(x);
				System.out.println(x);
				// 直接传递一个对象给Mysql
				// Mysql.insertInto(x);
			}
		});

		jsc.close();
	}
}
